﻿using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Text;
using System.Threading;
using Apache.NMS;
using log4net;
using Reactor.Environment;
using Reactor.Messaging.ServiceBus.Receiving;
using Samurai.Wakizashi;

namespace Reactor.Messaging.ServiceBus.Internals
{
    public partial class TaskDistributor
    {
        public class TaskProcessor : DisposableBase
        {
            #region Fields

            private static readonly ILog Log = LogManager.GetLogger(typeof(TaskProcessor));
            private readonly ConcurrentBag<IReceiver> _messageReceivers = new ConcurrentBag<IReceiver>();
            private readonly TaskDistributor _taskDistributor;
            protected readonly INmsGateway NmsGateway;
            protected ISession Session;
            private readonly BlockingCollection<TaskBase> _taskQueue;
            private Thread _executionThread;
            private const int MaxNumberOfTaskRetries = 3;

            #endregion

            #region Constructors

            /// <summary>
            /// Initializes a new instance of the <see cref="TaskProcessor"/> class.
            /// </summary>
            /// <param name="taskDistributor">The task distributor.</param>
            /// <param name="nmsGateway">The JMS manager.</param>
            public TaskProcessor(TaskDistributor taskDistributor, INmsGateway nmsGateway)
            {
                _taskDistributor = taskDistributor;
                NmsGateway = nmsGateway;
                _taskQueue = new BlockingCollection<TaskBase>();
            }

            #endregion

            /// <summary>
            /// Occurs when a message sent.
            /// </summary>
            public event EventHandler<MessageEventArgs> MessageSent;

            /// <summary>
            /// Occurs when a message is received.
            /// </summary>
            public event EventHandler<MessageEventArgs> MessageReceived;

            public virtual void Start()
            {
                Session = NmsGateway.GetSession();

                SetupDedicatedThread();
            }

            public void Stop()
            {
                _taskQueue.CompleteAdding();

                foreach (var messageReceiver in _messageReceivers)
                    messageReceiver.MessageReceived -= MessageReceiver_MessageReceived;
            }

            /// <summary>
            /// Processes the provided task. If an exception occurs while processing the task, it is returned to 
            /// the <see cref="TaskProcessor"/> for another enqueue attempt.
            /// </summary>
            /// <param name="task">The task.</param>
            public virtual void ProcessTask(TaskBase task)
            {
                // Assert that task is valid
                task.Validate();

                // Add task to blocking queue
                try
                {
                    _taskQueue.Add(task);
                }
                catch (InvalidOperationException)
                {
                     // IOE means that Take() or Add() was called on a complete collection
                     // In most cases, a harmless race condition during shutdown.
                }
            }

            /// <summary>
            /// Indicates whether this instance is servicing a particular MessagingChannel. 
            /// </summary>
            /// <param name="typedDestinationName">Typed MessagingChannel name.</param>
            /// <returns>
            /// 	<c>true</c> if [is servicing MessagingChannel] [the specified typed MessagingChannel name]; otherwise, <c>false</c>.
            /// </returns>
            public bool IsServicingDestination(string typedDestinationName)
            {
                return
                    (_messageReceivers.Any(messageReceiver => messageReceiver.MessagingChannel.GetTypedDestinationName() == typedDestinationName));
            }

            protected void SetupDedicatedThread()
            {
                _executionThread = new Thread(BeginServicingQueue) { Name = "Reactor ServiceBus TaskProcessor Thread" };
                _executionThread.Start();
            }

            private void BeginServicingQueue()
            {
                while (!_taskQueue.IsCompleted)
                {
                    TaskBase task = null;

                    try
                    {
                        task = _taskQueue.Take();
                    }
                    catch (InvalidOperationException)
                    {
                        // IOE means that Take() or Add() was called on a complete collection
                        // In most cases, a harmless race condition during shutdown.
                    }

                    if (task == null) continue;

                    if (task is TransmissionTask)
                        ProcessTransmissionTask((TransmissionTask)task);
                    else if (task is SubscriptionTask)
                        ProcessSubscriptionTask((SubscriptionTask)task);
                    else
                        throw new InvalidOperationException("Invalid task implementation. Only TransmissionTask and SubscriptionTask types are supported.");
                }
            }

            private void ProcessTransmissionTask(TransmissionTask transmissionTask)
            {
                try
                {
                    transmissionTask.ExecutionAttempts++;

                    Transmit(transmissionTask);
                }
                catch (Exception e)
                {
                    if(transmissionTask.ExecutionAttempts >= MaxNumberOfTaskRetries)
                        Log.Error("Unable to process transmission task and all retries have failed.", e);
                    else
                    {
                        Log.Error("Unable to process transmission task. Attempting to enqueue task again.", e);
                        _taskDistributor.EnqueueTask(transmissionTask);
                    }
                }
                finally
                {
                    BeginServicingQueue();
                }
            }

            private void ProcessSubscriptionTask(SubscriptionTask subscriptionTask)
            {
                try
                {
                    subscriptionTask.ExecutionAttempts++;

                    // Make sure we only create a single receiver per MessagingChannel
                    var receiver = _messageReceivers.SingleOrDefault(r => r.MessagingChannel.DestinationName == subscriptionTask.MessagingChannel.DestinationName);
                    if (receiver != null) return;

                    var jmsManager = Context.ServiceLocator.GetInstance<INmsGateway>();

                    // Create proper receiver
                    if (subscriptionTask.MessagingChannel is PubSubChannel)
                        _messageReceivers.Add(new SubscriptionReceiver(jmsManager, Session, subscriptionTask.MessagingChannel));
                    else
                        _messageReceivers.Add(new PointToPointChannelReceiver(jmsManager, Session, subscriptionTask.MessagingChannel));

                    // Start message receivers
                    foreach (var messageReceiver in _messageReceivers)
                    {
                        messageReceiver.MessageReceived += MessageReceiver_MessageReceived;
                        if (!messageReceiver.IsRunning) messageReceiver.Start();
                    }
                }
                catch (Exception e)
                {
                    if(subscriptionTask.ExecutionAttempts >= MaxNumberOfTaskRetries)
                        Log.Error("Unable to process subscription task and all retries have failed.", e);
                    else
                    {
                        Log.Error(string.Format("Unable to process subscription task. Attempting to place back on task queue."), e);
                        _taskDistributor.EnqueueTask(subscriptionTask);    
                    }
                }
                finally
                {
                    BeginServicingQueue();
                }
            }

            /// <summary>
            /// Handles the MessageReceived event of the PointToPointChannelReceiver control by raising the same event in turn.
            /// </summary>
            /// <param name="sender">The source of the event.</param>
            /// <param name="e">The <see cref="Reactor.Messaging.ServiceBus.MessageEventArgs"/> instance containing the event data.</param>
            protected void MessageReceiver_MessageReceived(object sender, MessageEventArgs e)
            {
                // Bubble MessageReceived event upward
                if (MessageReceived != null) MessageReceived(this, e);
            }

            private void Transmit(TransmissionTask task)
            {
                if (task == null) throw new ArgumentNullException("task");
                if(task.Message == null) throw new InvalidOperationException("Message to transmit not found in TransmissionTask");

                var jmsMessage = Session.CreateXmlMessage(task.Message);

                if(string.IsNullOrEmpty(jmsMessage.Text)) throw new Exception("Unable to serialize message prior to transmission.");

                IDestination jmsDestination = (task.MessagingChannel is PointToPointChannel)
                                      ? Session.GetQueue(task.MessagingChannel.DestinationName)
                                      : (IDestination)Session.GetTopic(task.MessagingChannel.DestinationName);
                EvaluateReplyInformation(task, jmsMessage);
                EvaluateWorkflowStatus(task, jmsMessage);

                using (var producer = Session.CreateProducer(jmsDestination))
                {
                    jmsMessage.IncrementNumberOfDeliveryAttempts();
                    producer.Send(jmsDestination, jmsMessage);
                }

                if (Log.IsDebugEnabled) Log.DebugFormat("Message of type: {0} was sent to {1}: {2}", task.Message.GetType().FullName, task.MessagingChannel.GetType().Name, task.MessagingChannel.DestinationName);
                
                // Raise message sent event if anyone is listening
                if (MessageSent == null) return;
                MessageSent(this, new MessageEventArgs { Message = task.Message, MessagingChannel = task.MessagingChannel });
            }

            private static void EvaluateWorkflowStatus(TransmissionTask task, ITextMessage jmsMessage)
            {
                // If this transmission task is tied to a workflow instance, do nothing
                if (task.WorkflowInstanceId == Guid.Empty) return;
                
                // Enhance the JMS message with the workflow id for correlation purposes.
                jmsMessage.NMSCorrelationID = task.WorkflowInstanceId.ToString();
            }

            private void EvaluateReplyInformation(TransmissionTask task, IMessage jmsMessage)
            {
                // If this transmission task is not a request, do nothing
                if (task.ReplyDestination == null) return;

                // Determine reply MessagingChannel and build appropriate jms string
                var sb = new StringBuilder(2);
                var replyDestinationName = (task.ReplyDestination is PointToPointChannel) 
                    ? sb.Append("queue://").Append(task.ReplyDestination.DestinationName).ToString()
                    : sb.Append("topic://").Append(task.ReplyDestination.DestinationName).ToString();

                // Now enhance the jms message with reply information
                jmsMessage.NMSReplyTo = Session.GetDestination(replyDestinationName); // task.RequestIdentifier.ToString());
                jmsMessage.Properties.SetString("RequestCorrelationId", task.RequestIdentifier.ToString());
            }

            protected override void Dispose(bool disposing)
            {
                if(_taskQueue != null) _taskQueue.Dispose();

                if(_messageReceivers != null)
                {
                    foreach (var messageReceiver in _messageReceivers)
                        messageReceiver.Dispose();
                }

                base.Dispose(disposing);
            }
        }
    }
}
